--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
Commit eff2edf7a1d53b493fef6adbda05283c1b0c3c01
Parents : e8f6a4e
Author : Ivan <ivan@quad4.io>
Signature : Signature validation error
Date : 2026-04-23T04:53:24-05:00
feat(tests): add property-based tests for Web Audio bridge PCM paths to ensure robustness and exception handling
Changes
1 files changed, 126 insertions(+), 0 deletions(-)
Diff
diff --git a/tests/backend/test_web_audio_bridge_hypothesis.py b/tests/backend/test_web_audio_bridge_hypothesis.py
new file mode 100644
index 00000000..2a2e3489
--- /dev/null
+++ b/tests/backend/test_web_audio_bridge_hypothesis.py
@@ -0,0 +1,126 @@
+# SPDX-License-Identifier: 0BSD
+
+"""Property-based tests for Web Audio bridge PCM paths (robustness, no uncaught exceptions)."""
+
+import asyncio
+from unittest.mock import MagicMock, patch
+
+import numpy as np
+import pytest
+from hypothesis import given, settings
+from hypothesis import strategies as st
+
+from meshchatx.src.backend.web_audio_bridge import (
+ WebAudioBridge,
+ WebAudioSink,
+ WebAudioSource,
+)
+
+
+class _DummySink:
+ def __init__(self):
+ self.frames = []
+
+ def can_receive(self, from_source=None):
+ return True
+
+ def handle_frame(self, frame, source):
+ self.frames.append(frame)
+
+
+@settings(max_examples=200, deadline=None)
+@given(pcm=st.binary(max_size=8192))
+def test_web_audio_source_push_pcm_never_propagates(pcm):
+ sink = _DummySink()
+ src = WebAudioSource(target_frame_ms=60, sink=sink)
+ with patch("meshchatx.src.backend.web_audio_bridge.RNS.log"):
+ src.push_pcm(pcm)
+
+
+@settings(max_examples=120, deadline=None)
+@given(pcm=st.binary(max_size=8192))
+def test_web_audio_bridge_push_client_frame_never_propagates(pcm):
+ bridge = WebAudioBridge(MagicMock(), MagicMock())
+ sink = _DummySink()
+ bridge.tx_source = WebAudioSource(target_frame_ms=60, sink=sink)
+ with patch("meshchatx.src.backend.web_audio_bridge.RNS.log"):
+ bridge.push_client_frame(pcm)
+
+
+@settings(max_examples=120, deadline=None)
+@given(pcm=st.binary(max_size=8192))
+def test_web_audio_source_push_pcm_sink_can_receive_false_never_propagates(pcm):
+ sink = MagicMock()
+ sink.can_receive.return_value = False
+ src = WebAudioSource(target_frame_ms=60, sink=sink)
+ with patch("meshchatx.src.backend.web_audio_bridge.RNS.log"):
+ src.push_pcm(pcm)
+ sink.handle_frame.assert_not_called()
+
+
+@settings(max_examples=150, deadline=None)
+@given(
+ values=st.lists(
+ st.floats(min_value=-4.0, max_value=4.0, allow_nan=False, allow_infinity=False),
+ max_size=2048,
+ ),
+ shape=st.sampled_from(["col", "row", "empty"]),
+)
+def test_web_audio_sink_numpy_frame_never_propagates(values, shape):
+ loop = asyncio.new_event_loop()
+ asyncio.set_event_loop(loop)
+ sent = []
+
+ async def _send_bytes(data):
+ sent.append(data)
+
+ sink = WebAudioSink(loop, _send_bytes)
+ if shape == "empty":
+ arr = np.zeros((0, 1), dtype=np.float32)
+ elif shape == "row":
+ arr = np.array(values, dtype=np.float32).reshape(1, -1) if values else np.zeros((1, 0), dtype=np.float32)
+ else:
+ arr = np.array(values, dtype=np.float32).reshape(-1, 1) if values else np.zeros((0, 1), dtype=np.float32)
+
+ with patch("meshchatx.src.backend.web_audio_bridge.RNS.log"):
+ sink.handle_frame(arr, None)
+ loop.run_until_complete(asyncio.sleep(0.05))
+ loop.close()
+
+
+@settings(max_examples=100, deadline=None)
+@given(raw=st.binary(max_size=1024))
+def test_web_audio_sink_bytes_frame_never_propagates(raw):
+ loop = asyncio.new_event_loop()
+ asyncio.set_event_loop(loop)
+ sent = []
+
+ async def _send_bytes(data):
+ sent.append(data)
+
+ sink = WebAudioSink(loop, _send_bytes)
+ with patch("meshchatx.src.backend.web_audio_bridge.RNS.log"):
+ sink.handle_frame(raw, None)
+ loop.run_until_complete(asyncio.sleep(0.05))
+ loop.close()
+
+
+@pytest.mark.asyncio
+@settings(max_examples=80, deadline=None)
+@given(
+ values=st.lists(
+ st.floats(min_value=-2.0, max_value=2.0, allow_nan=False, allow_infinity=False),
+ max_size=1024,
+ ),
+)
+async def test_web_audio_sink_running_loop_numpy_never_propagates(values):
+ sent = []
+
+ async def _send_bytes(data):
+ sent.append(data)
+
+ sink = WebAudioSink(asyncio.get_running_loop(), _send_bytes)
+ arr = np.array(values, dtype=np.float32).reshape(-1, 1) if values else np.zeros((0, 1), dtype=np.float32)
+ with patch("meshchatx.src.backend.web_audio_bridge.RNS.log"):
+ sink.handle_frame(arr, None)
+ await asyncio.sleep(0.05)
──────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────